package com.trt.beibei.manager.rocketmq;

import com.trt.beibei.bo.OrderMessageBO;
import com.trt.beibei.service.CouponBackService;
import com.trt.beibei.service.OrderCancelService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

// 消费者
// @Component
public class ConsumerSchedule implements CommandLineRunner { // CommandLineRunner：启动时候初始化资源，只执行一次

    @Value("${rocketmq.consumer.consumer-group}")
    private String consumerGroup;

    @Value("${rocketmq.namesrv-addr}")
    private String namesrvAddr;

    @Autowired
    private OrderCancelService orderCancelService;

    @Autowired
    private CouponBackService couponBackService;

    public void messageListener() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

        consumer.setNamesrvAddr(namesrvAddr);

        consumer.subscribe("TopicTest", "*");  // 订阅主题  子主题名称， *所有消息

        consumer.setConsumeMessageBatchMaxSize(1); // 每次消费几条消息

        // 注册消息的监听
        consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
            for (Message message : messages) {
                System.out.println("消息：" + new String(message.getBody()));
                byte[] body = message.getBody();
                String expiredKey = new String(body);

                OrderMessageBO messageBO = new OrderMessageBO(expiredKey);
                orderCancelService.cancel(messageBO);
                couponBackService.returnBack(messageBO);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
    }


    @Override
    public void run(String... args) throws Exception {
        messageListener();
    }
}
